package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.commom.base.BaseSqlApp;
import com.bw.gmall.realtime.commom.constant.Constant;
import com.bw.gmall.realtime.commom.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class CommentApp extends BaseSqlApp {
    public static void main(String[] args) {
        new CommentApp().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO, 4, 10012);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        /**
         * 1.通过flinksql把评论数据筛选出来
         * 2.读取码表数据，数据在Hbase
         * 3.评论跟码表进行join操作
         * 4、写出到kakfka主题
         */
//        tableEnv.ex`ecuteSql()
//        tableEnv.sqlQuery();
        // 1、读取ODS的数据
        readOdsDb(tableEnv,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
        // 2、过滤评论数据
        Table commentTab = getCommentTab(tableEnv);
        tableEnv.createTemporaryView("comment_info",commentTab);

        // 3、读取字典表
        readBaseDic(tableEnv);

        // 这用什么Join  left right  join  full  lookup
        // 用Lookup join需要用到处理时间,用大表和小表
        // 用Lookup join需要用到处理时间
        // 用Lookup join需要用到处理时间
        // 用Lookup join需要用到处理时间
        // 用Lookup join需要用到处理时间
        // 4、跟字典表进行Join操作
        Table commentJoin = getCommentJoin(tableEnv);
        // 5、创建对应的kafka表
        getTableResult(tableEnv);
        // 6、写出到Kafka
        commentJoin.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();

    }

    private TableResult getTableResult(StreamTableEnvironment tableEnv) {
        return tableEnv.executeSql("create table " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO + "(" +
                "  id STRING,\n" +
                "  user_id STRING,\n" +
                "  nick_name STRING,\n" +
                "  sku_id STRING,\n" +
                "  spu_id STRING,\n" +
                "  order_id STRING,\n" +
                "  appraise_code STRING,\n" +
                "  appraise_name STRING,\n" +
                "  comment_txt STRING,\n" +
                "  create_time STRING" +
                ")"
                + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }

    private Table getCommentJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("SELECT \n" +
                "  id ,\n" +
                "  user_id ,\n" +
                "  nick_name ,\n" +
                "  sku_id ,\n" +
                "  spu_id ,\n" +
                "  order_id ,\n" +
                "  appraise ,\n" +
                "  dic_name appraise_name,\n" +
                "  comment_txt ,\n" +
                "  create_time \n" +
                "FROM comment_info AS c\n" +
                "JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b\n" +
                "ON c.appraise = b.dic_code;");
    }

    private Table getCommentTab(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['nick_name'] nick_name,\n" +
                "  `data`['sku_id'] sku_id,\n" +
                "  `data`['spu_id'] spu_id,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['appraise'] appraise,\n" +
                "  `data`['comment_txt'] comment_txt,\n" +
                "   proc_time,\n" +
                "  `data`['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall' and `table` = 'commnet_info' and `type`='insert'");
    }
}
